package cn.xinfei.xdecision.data.core.util;

import cn.xinfei.xdecision.common.model.datax.datamanage.v2.VariableCallLog;
import cn.xinfei.xdecision.common.model.datax.exception.XDecisionDataException;
import cn.xinfei.xdecision.common.model.datax.exception.XDecisionDataExceptionType;
import cn.xinfei.xdecision.data.core.frame.executors.IExecutorOutput;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

public class VariableFutureUtils {

    private static final Logger LOGGER = LoggerFactory.getLogger(VariableFutureUtils.class);

    public static <T extends IExecutorOutput> Map<String, Object> flatFutureMap(List<T> futureReslut) {

        Map<String, Object> flatMap = Maps.newHashMap();
        if (CollectionUtils.isNotEmpty(futureReslut)) {
            flatMap = Optional.of(futureReslut).get()
                    .stream().filter(Objects::nonNull)
                    .map(T::executorReuslt)
                    .filter(Objects::nonNull)
                    .map(Map::entrySet).flatMap(Set::stream)
                    .filter(Objects::nonNull)
                    .collect(HashMap::new, (map, entry) -> map.put(entry.getKey(), entry.getValue()), HashMap::putAll);
        }
        return flatMap;
    }


    public static <T extends IExecutorOutput> List<VariableCallLog> flatCallLogs(List<T> futureReslut) {
        List<VariableCallLog> variableCallLogs = Lists.newArrayList();
        if (CollectionUtils.isNotEmpty(futureReslut)) {
            variableCallLogs = Optional.of(futureReslut).get()
                    .stream().filter(var -> Objects.nonNull(var) && CollectionUtils.isNotEmpty(var.varCallLogList()))
                    .map(T::varCallLogList).flatMap(Collection::stream)
                    .filter(Objects::nonNull)
                    .collect(Collectors.toList());
        }
        return variableCallLogs;
    }


    public static <T> List<T> getExecutorFutures(String prefix, List<Future<T>> futureList, long timeout) {
        if (CollectionUtils.isEmpty(futureList)) {
            LOGGER.info("{}_FUTURE_LIST IS EMPTY!", prefix);
            return Collections.emptyList();
        }
        List<T> result = Lists.newArrayList();
        futureList.forEach(future -> {
            try {
                T t = future.get(timeout, TimeUnit.SECONDS);
                result.add(t);
                LOGGER.info("future: {}", JSONObject.toJSONString(t));
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                if (LOGGER.isErrorEnabled()) {
                    LOGGER.error("error in GetExecutorFuture " + prefix + "_线程中断！", e);
                }
            } catch (ExecutionException e) {
                if (LOGGER.isErrorEnabled()) {
                    LOGGER.error("error in GetExecutorFuture " + prefix + "_业务异常: {}", e.getMessage(), e);
                }
            } catch (CancellationException | TimeoutException e) {
                if (LOGGER.isErrorEnabled()) {
                    LOGGER.error("error in GetExecutorFuture " + prefix + " future cancell or timeout :{}", e.getMessage(), e);
                }
            }
        });
        return result;
    }
}
